package cn.doitedu.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class _01_Demo_2_TableApi {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 构造一个源表定义描述器
        TableDescriptor tableDescriptor = TableDescriptor.forConnector("kafka")
                // 指定连接器
                .option("topic", "doit46")
                // 指定连接器参数
                .option("properties.bootstrap.servers", "doitedu:9092")
                .option("properties.group.id", "g003")
                .option("scan.startup.mode", "latest-offset")
                .option("value.format", "json")
                .option("value.fields-include", "EXCEPT_KEY")
                // 指定表结构
                .schema(Schema.newBuilder()
                        .column("uid", DataTypes.BIGINT())
                        .column("event_id", DataTypes.STRING())
                        .column("properties", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
                        .column("action_time", DataTypes.BIGINT())
                        .build())
                .build();

        // 利用表定义描述器创建一张表对象
        Table sourceTable = tenv.from(tableDescriptor);


        // 构造一个表描述，代表 目标表
        TableDescriptor sinkDescriptor = TableDescriptor.forConnector("kafka")
                // 指定连接器
                .option("topic", "page-load-events-2")
                // 指定连接器参数
                .option("properties.bootstrap.servers", "doitedu:9092")
                .option("properties.group.id", "g003")
                .option("scan.startup.mode", "latest-offset")
                .option("value.format", "json")
                .option("value.fields-include", "EXCEPT_KEY")
                // 指定表结构
                .schema(Schema.newBuilder()
                        .column("uid", DataTypes.BIGINT())
                        .column("event_id", DataTypes.STRING())
                        .column("properties", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
                        .column("action_time", DataTypes.BIGINT())
                        .build())
                .build();


        sourceTable
                .where(
                        $("event_id").isEqual("page_load")
                                .and($("properties").at("ref").isEqual("/x"))
                ).select($("uid"), $("event_id"), $("properties"), $("action_time"))
                .insertInto(sinkDescriptor)
                .execute();



    }


}
